//====================================================================
// Filename:      server.cc
//                
// It belongs to goblin-os.
//                
// Version:       
// Author:        Ins <InsFous@gmail.com>
// Created at:    Fri Dec 10 17:49:50 2010
// Description:   
//                
//====================================================================
#include "Queue.h"
#include "goblin_sqs_queue_center.h"
#include "goblin_sqs_queue_disk_sync_thread.h"

#include <signal.h>

#include <protocol/TBinaryProtocol.h>
#include <server/TSimpleServer.h>
#include <server/TThreadPoolServer.h>
#include <transport/TServerSocket.h>
#include <transport/TBufferTransports.h>
#include <concurrency/ThreadManager.h>
#include <concurrency/PosixThreadFactory.h>

using namespace ::apache::thrift;
using namespace ::apache::thrift::protocol;
using namespace ::apache::thrift::transport;
using namespace ::apache::thrift::server;
using namespace ::apache::thrift::concurrency;


using boost::shared_ptr;

using namespace goblin::sqs::thrift;

using goblin::sqs::QueueCenter;

class QueueHandler : virtual public QueueIf {
 public:
  QueueHandler() {
    // Your initialization goes here
    Init();
  }

  bool CreateQueue(const QueueName& queue_name, const QueueFlag::type queue_flag) {
    // Your implementation goes here
    printf("CreateQueue\n");
    return queue_center_->CreateQueue(queue_name, queue_flag);
  }

  void DeleteQueue(const QueueName& queue_name) {
    // Your implementation goes here
    printf("DeleteQueue\n");
    queue_center_->DeleteQueue(queue_name);
  }

  bool SendMessage(const QueueName& queue_name, const std::string& msg_content) {
    // Your implementation goes here
    printf("SendMessage\n");
    queue_center_->SendMessage(queue_name, msg_content, -1);
  }

  bool SendMessageWithType(const QueueName& queue_name, const std::string& msg_content, const int32_t msg_type) {
    // Your implementation goes here
    printf("SendMessageWithType\n");
    queue_center_->SendMessage(queue_name, msg_content, msg_type);
  }

  void FetchMessage(Message& _return, const QueueName& queue_name) {
    // Your implementation goes here
    printf("FetchMessage\n");
    queue_center_->FetchMessage(queue_name, _return);
  }

  void FetchMessageList(std::vector<Message> & _return, const QueueName& queue_name, const int32_t count) {
    // Your implementation goes here
    printf("FetchMessageList\n");
    queue_center_->FetchMessageList(queue_name, count, _return);
  }

  void GetQueueStatus(QueueStatus &_return, const QueueName& queue_name) {
    // Your implementation goes here
    printf("GetQueueStatus\n");
    queue_center_->GetQueueStatus(_return, queue_name);
  }

  bool SetQueueCapacity(const QueueName &queue_name, const int64_t capacity) {
    // Your implementation goes here
    printf("SetQueueCapacity\n");
    return queue_center_->SetQueueCapacity(queue_name, capacity);
  }

  void SetQueueFilter(const QueueName& queue_name, const int32_t msg_type) {
    // Your implementation goes here
    printf("SetQueueFilter\n");
    queue_center_->SetQueueFilter(queue_name, msg_type);
  }

  void RestoreQueueDefaultFilter(const QueueName& queue_name) {
    // Your implementation goes here
    printf("RestoreQueueDefaultFilter\n");
    queue_center_->RestoreQueueDefaultFilter(queue_name);
  }


  void Init() {
    printf("Initing...\n");
    // 1. Init The Main Queue Server
    queue_center_ = new QueueCenter(std::string("queue_db"));

  }
 public:
  QueueCenter *queue_center_;
};


static QueueHandler *queue_handler = NULL;

void SigHandler(int signu) {
  printf("Got Signal %d\n", signu);
  if (queue_handler)
    queue_handler->queue_center_->Save();
  exit(0);
}


int main(int argc, char **argv) {
  signal(SIGINT, SigHandler);
  int port = 9090;
  queue_handler = new QueueHandler();
  shared_ptr<QueueHandler> handler(queue_handler);
  shared_ptr<TProcessor> processor(new QueueProcessor(handler));
  shared_ptr<TServerTransport> serverTransport(new TServerSocket(port));
  shared_ptr<TTransportFactory> transportFactory(new TBufferedTransportFactory());
  shared_ptr<TProtocolFactory> protocolFactory(new TBinaryProtocolFactory());

  shared_ptr<PosixThreadFactory> threadFactory(new PosixThreadFactory());
  shared_ptr<ThreadManager> threadManager(ThreadManager::newSimpleThreadManager(8));
  threadManager->threadFactory(threadFactory);
  threadManager->start();

  pthread_t tid;
  pthread_attr_t attr;
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
  pthread_create(&tid, &attr, goblin::sqs::queue_disk_sync_thread, queue_handler->queue_center_);
  pthread_attr_destroy(&attr);

  TThreadPoolServer server(processor, serverTransport, transportFactory, protocolFactory, threadManager);
  server.serve();

  return 0;
}

